package Autodesk;

import java.io.IOException;
import java.util.ArrayList;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

import DatabaseManaging.JdbcProductsDao;

public class ConceptTest extends Configured implements Tool  {
	
	public Path prodPath;
	public static class LogMapper extends Mapper<Object, Text, Text, IntWritable> {
		
		private final static IntWritable one = new IntWritable(1);
		private Text word = new Text("b");

	public void map(Object key, Text value, Context context)
			throws IOException, InterruptedException {
		System.out.println(word+" value: "+value.toString());
		context.write(new Text("a"), one);
	}
}

public static class LogReducer extends
		Reducer<Text, IntWritable, Text, IntWritable> {
	
	private IntWritable result = new IntWritable(1);

	public void reduce(Text key, Iterable<IntWritable> values,
			Context context) throws IOException, InterruptedException {
//	   	DriverManagerDataSource dataSource = new DriverManagerDataSource();
//	   	dataSource.setDriverClassName("com.mysql.jdbc.Driver");
//	   	dataSource.setUrl("jdbc:mysql://localhost:3306/ea");
//	   	dataSource.setUsername("root");
//	   	dataSource.setPassword("");
//    	JdbcProductsDao dao=new JdbcProductsDao();
//    	dao.setDataSource(dataSource);   	
		ApplicationContext cont = new ClassPathXmlApplicationContext("application-context-persist.xml");
	    JdbcProductsDao dao=(JdbcProductsDao) cont.getBean("JdbcProductsDao");
		ArrayList<Object> list=dao.findElementby("Autocad");
		System.out.println(list.toString());
		for (Object val : list) {
			context.write(new Text(val.toString()), result);
		}	
		
	}
}

public int run(String[] args) throws Exception { 
	Configuration conf = getConf();
	
	Job job = new Job(conf, "ParseLogs");
	job.setJarByClass(ConceptTest.class);

	job.setMapperClass(LogMapper.class);
	job.setReducerClass(LogReducer.class);
	
	job.setOutputKeyClass(Text.class);
	job.setOutputValueClass(IntWritable.class);

    FileInputFormat.setInputPaths(job, new Path(args[0]));
    FileOutputFormat.setOutputPath(job, new Path(args[1]) );

	System.exit(job.waitForCompletion(true)?0:1);
	
	return 0;

}

public static void main(String[] args) throws Exception {
	int res = ToolRunner.run(new Configuration(), new ConceptTest(), args);
	System.exit(res);
	}


}